package com.hoolai.qps.netty.proto;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;

import com.hoolai.qps.codes.netty.ClientMsg;
import com.hoolai.qps.codes.netty.LengthFieldBasedFrameDecoder;

public class ProxyServer {

    private final ChannelGroup clients = new DefaultChannelGroup();

    private final ConcurrentHashMap<Integer, Channel> servers = new ConcurrentHashMap<Integer, Channel>(3);

    private String hostName;

    public static void main(String[] args) {
        ProxyServer proxyServer = new ProxyServer();
        proxyServer.start("127.0.0.1");
    }

    public void start(String hostName) {
        this.hostName = hostName;
        logicProxyServerStart();
        clientProxyServerStart();
        System.err.println("proxy server start!");
    }

    private void logicProxyServerStart() {
        ServerBootstrap bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors
                .newCachedThreadPool(), Executors.newCachedThreadPool()));
        final ServerProxyHandler handler = new ServerProxyHandler();
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(new LengthFieldBasedFrameDecoder(81920, 0, 4, 0, 4, false), handler);
            }
        });
        bootstrap.setOption("child.tcpNoDelay", true);
        bootstrap.setOption("child.keepAlive", true);
        bootstrap.bind(new InetSocketAddress(hostName, 1111));

        handler.showCounter("server side:");
    }

    private void clientProxyServerStart() {
        ServerBootstrap bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors
                .newCachedThreadPool(), Executors.newCachedThreadPool()));
        final ClientProxyHandler handler = new ClientProxyHandler();
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(new LengthFieldBasedFrameDecoder(81920, 0, 4, 0, 4, false),
                        new ClientProxyDecoder(), handler);
            }
        });
        bootstrap.setOption("child.tcpNoDelay", true);
        bootstrap.setOption("child.keepAlive", true);
        bootstrap.bind(new InetSocketAddress(hostName, 2222));

        handler.showCounter("client side:");
    }

    private class ClientProxyHandler extends CounterHandler {

        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            counter.incrementAndGet();
            ClientMsg clientMsg = (ClientMsg) e.getMessage();
            ChannelBuffer channelBuffer = clientMsg.getChannelBuffer();
            channelBuffer.setInt(8, ctx.getChannel().getId());// set channel id
            notifyLogicServer(clientMsg.getCmdId(), channelBuffer);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
            System.err.println("proxy exception caught:" + e.getCause());
            ProxyServer.this.clients.remove(ctx.getChannel().getId());
            ctx.getChannel().close();
        }

        @Override
        public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
            System.err.println("close");
            ProxyServer.this.clients.remove(ctx.getChannel());
            ctx.getChannel().close();
        }

        @Override
        public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
            clients.add(ctx.getChannel());
        }

    }

    private class ClientProxyDecoder extends OneToOneDecoder {
        @Override
        protected Object decode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
            ChannelBuffer buf = (ChannelBuffer) msg;
            int cmdId = buf.getInt(4);
            ClientMsg clientMsg = new ClientMsg(cmdId, buf);
            return clientMsg;
        }
    }

    private abstract class CounterHandler extends SimpleChannelUpstreamHandler {
        protected AtomicInteger counter = new AtomicInteger();

        private volatile int last = 0;

        public int getCounter() {
            int result = counter.get() - last;
            last = counter.get();
            return result;
        }

        public void showCounter(final String message) {
            Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    System.err.println(message + CounterHandler.this.getCounter() + " ,time="
                            + System.currentTimeMillis());
                }
            }, 0l, 1l, TimeUnit.SECONDS);
        }
    }

    private class ServerProxyHandler extends CounterHandler {
        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            counter.incrementAndGet();
            ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
            if (isRegisterCmd(buffer)) {
                registerLogicServer(ctx.getChannel(), buffer.getInt(8));// register logic server
                System.err.println("register succ server id=" + buffer.getInt(8));
            } else {
                int channelId = buffer.getInt(8);
                Channel channel = clients.find(channelId);
                if (null != channel) {
                    channel.write(buffer);// send buffer to client
                }
            }
        }
    }

    private boolean isRegisterCmd(ChannelBuffer buffer) {
        return buffer.getInt(4) == 0;
    }

    public void registerLogicServer(Channel channel, int serverId) {
        servers.put(serverId, channel);
    }

    public void notifyLogicServer(int cmdId, ChannelBuffer channelBuffer) {
        cmdId = Integer.valueOf(String.valueOf(cmdId).subSequence(0, 3).toString());//TODO $
        Channel channel = servers.get(cmdId);
        if (null != channel) {
            channel.write(channelBuffer);
        }
    }

}
